/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2024-2024. All rights reserved.
 * Description: 异构集合通信CPU侧event
 */
#include "esched_notify.h"
#include "dlhal_function.h"
#include "sal_pub.h"

namespace hccl {
constexpr u32 WAIT_LOOP_TIME_AS_MILLI_SECOND = 10;
constexpr u32 MULTIPLIER_S2MS = 1000;

constexpr u32 DEFAULT_NTC_EVENTID = 50;
constexpr u32 MAX_THREAD_NUM = 512;
constexpr u32 MAX_THREAD_ID = 1024;
constexpr u32 NEVER_TIMEOUT = 0xffffffff; // 永不超时

std::mutex g_queThreadMutex;
std::queue<uint32_t> g_queThread;
static bool g_hasInited = false;

uint32_t EschedNotify::eventId_ = DEFAULT_NTC_EVENTID;
uint32_t EschedNotify::initialThreadId_ = 0U;
uint32_t EschedNotify::groupId_ = 0U;

EschedNotify::EschedNotify(NotifyType notifyType) : NotifyBase(notifyType)
{
}

EschedNotify::EschedNotify(NotifyType notifyType, HcclNotifyInfo notifyInfo)
    : NotifyBase(notifyType, notifyInfo)
{
}

EschedNotify::~EschedNotify()
{
    (void)Destroy();
}

HcclResult EschedNotify::Open()
{
    HCCL_DEBUG("[EschedNotify][Open] ntcPid[%u],ntcGrpId[%u], ntcTid[%u], ntcSubEventId[%u], ntcEventId[%u], "
        "msgLen[%u], phyDevId[%u], notify type[%u].", notifyInfo_.eschedEvent.ntcPid, notifyInfo_.eschedEvent.ntcGrpId,
        notifyInfo_.eschedEvent.ntcTid,  notifyInfo_.eschedEvent.ntcSubEventId, notifyInfo_.eschedEvent.ntcEventId,
        notifyInfo_.eschedEvent.msgLen, notifyInfo_.eschedEvent.phyDevId, notifyInfo_.type);
    return HCCL_SUCCESS;
}

HcclResult EschedNotify::Close()
{
    return HCCL_SUCCESS;
}

HcclResult EschedNotify::Wait(Stream& stream, HcclDispatcher dispatcher, s32 stage, u32 timeOut)
{
    struct event_info event;
    // 确认事件调度超时时间单位
    HcclResult ret = HCCL_SUCCESS;
    if (timeOut == 0) {
        ret = hrtHalEschedWaitEvent(sendinfo_.devId, sendinfo_.grpId, sendinfo_.threadId,
            timeOut, &event);
        if (ret != HCCL_E_AGAIN && ret != HCCL_SUCCESS) {
            HCCL_ERROR("hrtHalEschedWaitEvent fail. ret[%u] curPid[%d] curTid[%d] devId[%u] grpId[%u] threadId[%u]",
             ret, SalGetPid(), SalGetTid(), sendinfo_.devId, sendinfo_.grpId, sendinfo_.threadId);
        }
        return ret;
    } else if (timeOut == NEVER_TIMEOUT) {
        break_ = false;
        isWaiting_ = true;
        while (true) {
            ret = hrtHalEschedWaitEvent(sendinfo_.devId, sendinfo_.grpId, sendinfo_.threadId,
                WAIT_LOOP_TIME_AS_MILLI_SECOND, &event);
            if (ret == HCCL_E_AGAIN && break_) {
                break_ = false;
                isWaiting_ = false;
                HCCL_RUN_WARNING("hrtHalEschedWaitEvent break.");
                return HCCL_E_AGAIN;
            } else if (ret == HCCL_SUCCESS) {
                break;
            }
        }
        isWaiting_ = false;
    } else {
        u64 time = 0;
        break_ = false;
        isWaiting_ = true;
        while (time <= timeOut * MULTIPLIER_S2MS) {
            ret = hrtHalEschedWaitEvent(sendinfo_.devId, sendinfo_.grpId, sendinfo_.threadId,
                WAIT_LOOP_TIME_AS_MILLI_SECOND, &event);
            time += WAIT_LOOP_TIME_AS_MILLI_SECOND;
            if (ret == HCCL_E_AGAIN && break_) {
                break_ = false;
                isWaiting_ = false;
                return HCCL_E_AGAIN;
            } else if (ret == HCCL_SUCCESS) {
                break;
            }
        }
        if (time > timeOut * MULTIPLIER_S2MS && ret != HCCL_SUCCESS) {
            isWaiting_ = false;
            return HCCL_E_AGAIN;
        }
        isWaiting_ = false;
    }

    return HCCL_SUCCESS;
}

HcclResult EschedNotify::Post(Stream& stream, HcclDispatcher dispatcher, s32 stage)
{
    CHK_RET(hrtIpcIntNotice(&notifyInfo_.eschedEvent, stream.ptr()));
    return HCCL_SUCCESS;
}

HcclResult EschedNotify::Wait(Stream& stream, HcclDispatcher dispatcher, s32 stage, u32 timeOut,
    u32 userRank, u32 remoteUserRank)
{
    return Wait(stream, dispatcher, stage, timeOut);
}

HcclResult EschedNotify::Post(Stream& stream, HcclDispatcher dispatcher, s32 stage, u32 remoteUserRank)
{
    return Post(stream, dispatcher, stage);
}

HcclResult EschedNotify::SetIpc()
{
    s32 psPid = 0;
    hrtDrvDeviceGetBareTgid(psPid);
    notifyInfo_.eschedEvent.ntcPid = psPid;
    notifyInfo_.eschedEvent.ntcGrpId = groupId_;
    notifyInfo_.eschedEvent.ntcEventId = eventId_; // DEFAULT_NTC_EVENTID
    notifyInfo_.eschedEvent.ntcSubEventId = 0;
    notifyInfo_.eschedEvent.ntcTid = sendinfo_.threadId;
    notifyInfo_.eschedEvent.msgLen = 0;
    s32 logicDevId = 0;
    u32 phyDevId = 0;
    CHK_RET(hrtDrvDeviceGetPhyIdByIndex(logicDevId, phyDevId));
    notifyInfo_.eschedEvent.phyDevId = static_cast<u16>(phyDevId);
    HCCL_DEBUG("EschedNotify SetIpc ntcPid[%u],ntcGrpId[%u], ntcTid[%u], ntcSubEventId[%u], ntcEventId[%u], "
        "msgLen[%u], phyDevId[%u].", notifyInfo_.eschedEvent.ntcPid, notifyInfo_.eschedEvent.ntcGrpId,
        notifyInfo_.eschedEvent.ntcTid,  notifyInfo_.eschedEvent.ntcSubEventId, notifyInfo_.eschedEvent.ntcEventId,
        notifyInfo_.eschedEvent.msgLen, notifyInfo_.eschedEvent.phyDevId);

    return HCCL_SUCCESS;
}

HcclResult EschedNotify::Grant(s64 recvId)
{
    return HCCL_SUCCESS;
}

HcclResult EschedNotify::Alloc()
{
    uint32_t eventId = eventId_; // event_id可配
    HcclResult ret = HCCL_SUCCESS;
    uint64_t eventBitMap = static_cast<uint64_t>(1) << static_cast<uint64_t>(eventId);
    // 生成 thread id
    // host侧只有一个device,deviceid固定写死为0
    sendinfo_.devId = 0;
    sendinfo_.grpId = groupId_;
    sendinfo_.eventBitMap = eventBitMap;
    ret = ThreadIdCreate(sendinfo_.threadId);
    if (ret != HCCL_SUCCESS) {
        HCCL_ERROR("[EschedNotify][Alloc] get threadid fail");
        return ret;
    }

    ret = hrtHalEschedSubscribeEvent(sendinfo_.devId, sendinfo_.grpId, sendinfo_.threadId, sendinfo_.eventBitMap);
    if (ret != HCCL_SUCCESS) {
        HCCL_ERROR("[EschedNotify][Alloc]hrtHalEschedSubscribeEvent fail");
        return ret;
    }

    struct event_info event;
    ret = hrtHalEschedWaitEvent(sendinfo_.devId, sendinfo_.grpId, sendinfo_.threadId, 0, &event);
    if (ret != HCCL_E_AGAIN) {
        HCCL_ERROR("[EschedNotify][Alloc]hrtHalEschedWaitEvent fail");
        return ret;
    }
    HCCL_INFO("EschedNotify Alloc group id %u, event id:%u, thread id %d.",
        sendinfo_.grpId, eventId, sendinfo_.threadId);

    return HCCL_SUCCESS;
}

HcclResult EschedNotify::Destroy()
{
    if (this->sendinfo_.threadId != INVALID_UINT) {
        Break();
        std::unique_lock<std::mutex> lock(g_queThreadMutex);
        g_queThread.push(this->sendinfo_.threadId);
        HCCL_INFO("EschedNotify push que %u", this->sendinfo_.threadId);
        this->sendinfo_.threadId = INVALID_UINT;
    }
    return HCCL_SUCCESS;
}

HcclResult EschedNotify::ThreadIdCreate(uint32_t &threadId)
{
    std::unique_lock<std::mutex> lock(g_queThreadMutex);
    if (g_hasInited) {
        if (g_queThread.empty()) {
            HCCL_ERROR("[EschedNotify][ThreadIdCreate] ThreadIdCreate limited, limit value[%u]", MAX_THREAD_NUM);
            return HCCL_E_INTERNAL;
        }
    } else {
        HCCL_ERROR("[EschedNotify][ThreadIdCreate] g_queThread does not have Inited!");
        return HCCL_E_NOT_FOUND;
    }

    threadId = g_queThread.front();
    g_queThread.pop();
    return HCCL_SUCCESS;
}

HcclResult EschedNotify::InitGroupId()
{
    CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
    CHK_RET(hrtHalEschedAttachDevice(0));

    HcclResult ret = GetGroupId();
    if (ret == HCCL_SUCCESS) {
        HCCL_DEBUG("[EschedNotify][InitGroupId] grp has Inited!, groupid [%u]", groupId_);
        return HCCL_SUCCESS;
    }
    CHK_RET(hrtHalEschedCreateGrpEx(0, &groupId_, MAX_THREAD_ID, GRP_TYPE_BIND_CP_CPU));

    return HCCL_SUCCESS;
}

void EschedNotify::ThreadIdQueInit()
{
    std::unique_lock<std::mutex> lock(g_queThreadMutex);
    if (g_hasInited == true) {
        HCCL_DEBUG("[EschedNotify][ThreadIdQueInit] g_queThread has Inited! cannot repeat inited!");
        return;
    }

    if (InitGroupId() != HCCL_SUCCESS) {
        return;
    }

    if (g_queThread.empty()) {
        for (uint32_t i = 0; i < MAX_THREAD_NUM; i++) {
            g_queThread.push(i + initialThreadId_);
        }
        HCCL_DEBUG("g_queThread[%p] push success!", &g_queThread);
    }
    g_hasInited = true;
    lock.unlock();
    HCCL_DEBUG("[EschedNotify][ThreadIdQueInit] g_queThread Inited!!!");
    return;
}

HcclResult EschedNotify::GetGroupId()
{
    if (groupId_ != 0) {
        HCCL_DEBUG("[EschedNotify][GetGroupId] group id has set");
        return HCCL_SUCCESS;
    }

    struct esched_query_gid_input queryInput;
    queryInput.pid = SalGetPid();
    char groupName[HCCL_DRIVER_GROUP_NAME] = {"hcclGrpName"};
    s32 memRet = memcpy_s(queryInput.grp_name, sizeof(groupName), groupName, sizeof(groupName));
    CHK_PRT_RET(memRet != HCCL_SUCCESS, HCCL_ERROR("[halEschedCreateGrpEx] memcpy fail,"
        "return[%d].", memRet), HCCL_E_MEMORY);

    struct esched_input_info inPut;
    inPut.inBuff = reinterpret_cast<void *>(&queryInput);
    inPut.inLen = sizeof(queryInput);

    struct esched_output_info outPut;
    struct esched_query_gid_output queryOutput;
    outPut.outBuff = reinterpret_cast<void *>(&queryOutput);
    outPut.outLen = sizeof(queryOutput);

    memRet = hrtEschedQueryInfo(0, QUERY_TYPE_LOCAL_GRP_ID, &inPut, &outPut);
    CHK_PRT_RET(memRet != HCCL_SUCCESS, HCCL_WARNING("[GetGroupId] hrtEschedQueryInfo fail, maybe not inited,"
        "return[%d].", memRet), HCCL_E_DRV);

    groupId_ = reinterpret_cast<struct esched_query_gid_output *>(outPut.outBuff)->grp_id;
    HCCL_DEBUG("[EschedNotify][GetGroupId] grp id is %u.", groupId_);
    return HCCL_SUCCESS;
}

void EschedNotify::Break()
{
    if (isWaiting_) {
        break_ = true;
        while (isWaiting_) {
            break_ = true;
            SaluSleep(ONE_HUNDRED_MICROSECOND_OF_USLEEP);
        }
    }
}
}